//! @Author: DengLibin
//! @Date: Create in 2025-01-20 16:07:27
//! @Description:

use my_util_lib::date;
use rocketmq::conf::{ClientOption, SimpleConsumerOption};
use rocketmq::model::common::{FilterExpression, FilterType};
use rocketmq::SimpleConsumer;


pub async fn demo_receive() {
    // It's recommended to specify the topics that applications will publish messages to
    // because the simple consumer will prefetch topic routes for them on start and fail fast in case they do not exist
    let mut consumer_option = SimpleConsumerOption::default();
    consumer_option.set_topics(vec!["deng"]);
    consumer_option.set_consumer_group("SimpleConsumerGroup");

    // set which rocketmq proxy to connect
    let mut client_option = ClientOption::default();
    client_option.set_access_url("localhost:8081");
    client_option.set_enable_tls(false);

    // build and start simple consumer
    let mut consumer = SimpleConsumer::new(consumer_option, client_option).unwrap();
    let start_result = consumer.start().await;
    if start_result.is_err() {
        eprintln!(
            "simple consumer start failed: {:?}",
            start_result.unwrap_err()
        );
        return;
    }

    // pop message from rocketmq proxy
    let receive_result = consumer
        .receive(
            "deng".to_string(),
            // &FilterExpression::new(FilterType::Tag, "test_tag"),
            // &FilterExpression::new(FilterType::Tag, "test_tag_2"),
            &FilterExpression::new(FilterType::Tag, "tag"),
        )
        .await;
    if receive_result.is_err() {
        eprintln!("receive message failed: {:?}", receive_result.unwrap_err());
        return;
    }

    let messages = receive_result.unwrap();

    if messages.is_empty() {
        println!("no message received");
        return;
    }

    for message in messages {
        println!("收到消息,born_host:{}, body:{}, keys:{}, message_id:{}, namespace:{}, topic:{}, born_time:{}",
                 message.born_host(),
                 String::from_utf8_lossy(message.body()),
                 message.keys().join(","),
                 message.message_id(),
                 message.namespace(),
                 message.topic(),
                 date::timestamp2datetime(message.born_timestamp()).unwrap());
        // ack message to rocketmq proxy
        let ack_result = consumer.ack(&message).await;
        if ack_result.is_err() {
            eprintln!(
                "ack message {} failed: {:?}",
                message.message_id(),
                ack_result.unwrap_err()
            );
        }
    }

    // shutdown the simple consumer when you don't need it anymore.
    // you should shutdown it manually to gracefully stop and unregister from server
    let shutdown_result = consumer.shutdown().await;
    if shutdown_result.is_err() {
        eprintln!(
            "simple consumer shutdown failed: {:?}",
            shutdown_result.unwrap_err()
        );
    }
}